#!/usr/bin/perl

# Author: Lipyeow Lim
# Date  : 12/21/2010

use IO::Socket;
use IO::Select;
use threads;
use Data::Dumper;
use Time::HiRes qw(gettimeofday tv_interval);
use Log::Fast;


use strict;

scalar @ARGV==1 or 
   die "Usage: execplan.pl <plan file>\n";

my $log = Log::Fast->new({
         level => 'DEBUG',
         prefix => '%L %F:',
         type => 'fh',
         fh => \*STDERR,
        });

open( PLAN, "<$ARGV[0]") or 
   die "Cannot open plan file $ARGV[0]!\n"; 
my @lines = <PLAN>;
close PLAN;

my $plan;

eval join(" ",@lines);

(defined $plan) or die "Unable to read plan\n";

execQEP($plan);


#------------------------------------------------
#------------------------------------------------
sub execQEP
{
   my ($plan) = @_;

   my $t0 = [gettimeofday];
   #print join(" ", findSink($plan)), "\n"; exit;
   my ($sink, $ns) = findSink($plan);

   $log->DEBUG("# Result Stream = $ns");

   my $rsock= listenForResults($sink, $ns);

   foreach my $job (@$plan)
   {
      my $whost = $job->[0];
      sendJob($whost, $job->[1]);
   }


   # simple tests seem to show that IO::select is more
   # efficient than the multithread method
   #my $cnt = recvResults($rsock, $ns);
   my $cnt = recvResults2($rsock, $ns);

   print "cardinality = $cnt tuples\n";
   print "time = ", tv_interval($t0), "s \n";
   close $rsock;
}

#------------------------------------------------
# assumes input is a single sink streaming QEP
# finds the sink hostname:port and the number of
# streams feeding the sink.
#------------------------------------------------
sub findSink
{
   my ($qep) = @_;

   my %outhosts = ();
   my %ophosts= ();
   foreach my $op (@$qep)
   {
      my $curhost = $op->[0];
      $ophosts{$curhost}++;
      foreach my $outhost (@{$op->[1]->{'output'}})
      {
         $outhosts{$outhost}++;
      } 
   }

   # a sink is a node that is an output but not running any
   # operator by definition of a streaming QEP.
   #
   #print "outhost:",  Dumper(\%outhosts);
   #print "ophost:", Dumper(\%ophosts);
   my $sinkhost = "";
   foreach my $outhost (keys %outhosts )
   {
      if ( ! exists $ophosts{$outhost} )
      {
         $sinkhost=$outhost;
         last;
      }
   }
   return ($sinkhost, $outhosts{$sinkhost});
}

#------------------------------------------------
#------------------------------------------------
sub listenForResults 
{
   my ($hostport, $ns) = @_;

   my $sock = new IO::Socket::INET ( LocalHost => $hostport,
                                  Proto => 'tcp',
                                  Listen => $ns,
                                  ReuseAddr => 1
                          );

   die "Could not create socket on port $hostport: $!\n" unless $sock;

   return $sock;
}


#------------------------------------------------
#------------------------------------------------
sub recvResults 
{
   my ($sock, $ns) = @_;
   my $clientsock;
   my @threads = ();
   for( my $i=0; $i<$ns; $i++) 
   {
      $clientsock = $sock->accept();
      my $t = threads->new( \&dothread, $clientsock );
      push @threads, $t;
   }

   foreach (@threads)
   {
      $_->join();
   }
   close $sock;
   return "unknown";
}

#------------------------------------------------
# this version uses IO::Select instead of threads 
# to do the union reads from multiple sockets
#------------------------------------------------
sub recvResults2 
{
   my ($sock, $ns) = @_;
   my $clientsock;
   my $sel = new IO::Select;
   my $count = 0;

   # if single result stream
   if( $ns==1 )
   {
      $clientsock = $sock->accept();
      while( <$clientsock> )
      {
         print $_;
         $count++;
      }
      return $count;
   }

   # if multiple result stream, do an implicit union
   for( my $i=0; $i<$ns; $i++) 
   {
      $clientsock = $sock->accept();
      $sel->add($clientsock);
   }

   while(my @ready = $sel->can_read())
   {
      foreach my $fh (@ready)
      {
         my $row = <$fh>;
         print $row;
         $count++;
         if($fh->eof)
         {
            $sel->remove($fh);
         }
      }
   }
   return $count;
}

#------------------------------------------------
#------------------------------------------------
sub dothread {
   my ($clientsock) = @_;
   my $tid = threads->tid();
   $log->DEBUG(" $tid executing ...\n");

   while (<$clientsock>)
   {
      print "$tid:$_";
   }
   close $clientsock;
}

#------------------------------------------------
#------------------------------------------------
sub sendJob
{
   my ($hostport,$msg) = @_;

   my $sock = new IO::Socket::INET ( 
                        PeerAddr => $hostport,
                        Proto => 'tcp');

   die "sendJob could not create socket : $!\n" unless $sock;

   print $sock Data::Dumper->Dump([$msg], ['msg']);

   close $sock;
}
